package com.swak.paxos.store;

import java.io.File;
import java.io.IOException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.mmap.MappedEndlessFile;
import com.swak.mmap.MappedFileAllocator;
import com.swak.mmap.MappedFileNaming;
import com.swak.paxos.PaxosConfig;
import com.swak.serializer.KryoSerializer;
import com.swak.serializer.Serializer;

/**
 * PaxosStorage.
 * 
 * @author 618lf
 */
public class PaxosStorage {

	private static Logger logger = LoggerFactory.getLogger(PaxosStorage.class);
	private static final Serializer Serializer = new KryoSerializer();

	private final String storePath;
	private final PaxosConfig config;
	private final MappedEndlessFile storage;

	public PaxosStorage(int groupID, PaxosConfig config, MappedFileAllocator mappedFileAllocator) {
		this.config = config;
		this.storePath = this.config.getPaxosLogStoragePath() + File.separator + groupID;
		this.storage = new MappedEndlessFile(this.storePath, new MappedFileNaming("edits_", ".dat"),
				this.config.getPaxosLogFileSize(), this.config.getPaxosLogRecordSize(),
				this.config.getFlushPhysicLogLeastPages(), mappedFileAllocator);
	}

	/**
	 * 初始化
	 */
	public void init() {

	}

	/**
	 * 存储数据.
	 * 
	 * @param log
	 * @throws IOException
	 */
	public boolean storage(PaxosLog log) {
		try {
			this.storage.appendData(PaxosStorage.encode(log));
			return true;
		} catch (Exception e) {
			logger.error("storage PaxosLog error.", e);
		}
		return false;
	}

	public void close() {
		try {
			this.storage.shutdown(1000);
		} catch (Exception e) {
		}
	}

	/**
	 * encode log.
	 * 
	 * @param log
	 * @return
	 */
	protected static byte[] encode(PaxosLog log) {
		return Serializer.serialize(log);
	}

	/**
	 * decode log.
	 * 
	 * @param log
	 * @return
	 */
	protected static PaxosLog decode(byte[] bytes) {
		return Serializer.deserialize(bytes);
	}
}